﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;

namespace Redis订阅模式DEMO
{
    public class Redis数据订阅
    {



        public string 实时数据通知名 = "INREAL";

        public bool IsCanRuning = true;
        /// <summary>
        /// Redis联接
        /// </summary>
        StackExchange.Redis.ConnectionMultiplexer rdidsConn = null;
        Dictionary<long, StackExchange.Redis.ISubscriber> subs = new Dictionary<long, StackExchange.Redis.ISubscriber>();

        /// <summary>
        /// 接收发布数据的方法 
        /// </summary>
        /// <param name="thrnum">生成的线程数据</param>
        /// <param name="speed">最高生成速度</param>
        public void 接收订阅数据(int thrnum, int speed)
        {
            InitConn();



            //获取发布订阅对像
            var sub = rdidsConn.GetSubscriber();
            for (int i = 0; i < thrnum; i++)
            {
                var ssus = rdidsConn.GetSubscriber();
                subs.Add(i, ssus);
            }

            接收订阅数据1(thrnum, speed);


            while (IsCanRuning)
            {

                记数();
            }

        }

        protected void InitConn()
        {
            try
            {
                rdidsConn = StackExchange.Redis.ConnectionMultiplexer.Connect("192.168.2.110:6379");
            }
            catch (Exception ex) { Console.WriteLine(ex); }
        }
        //产生模拟数据
        string demoStr = string.Join(",", System.Linq.Enumerable.Range(5000, 500).Select(p => p.ToString("d4")).ToArray());

        /// <summary>
        /// 记数器
        /// </summary>
        long llong = 0;
        /// <summary>
        /// 上次的记录数
        /// </summary>
        long perlong = 0;

        protected void 接收订阅数据1(int thrnum, int speed)
        {
            //判断Redis联接是否有效,异常处理
            if (rdidsConn != null && rdidsConn.IsConnected)
            {

                //并行发布实时数据
                System.Threading.Tasks.Parallel.For(0, thrnum, new ParallelOptions() { MaxDegreeOfParallelism = thrnum }, (n) =>
                {
                    try
                    {
                        var csub = subs[n % thrnum];
                        //联接断开时直接返回
                        if (!csub.IsConnected()) { return; }
                        csub.Subscribe(实时数据通知名, ReceiveData);
                    }
                    //联接断开异常
                    catch (StackExchange.Redis.RedisConnectionException) { }
                    catch (System.TimeoutException) { }
                });
            }
            else
            {
                if (rdidsConn == null)
                {
                    InitConn();
                }
            }
        }

        void 记数()
        {
            DateTime dt = DateTime.Now;
            long speed = llong - perlong;
            perlong = llong;

            if (rdidsConn == null)
            {
                InitConn();
            }
            //时间差毫秒
            double ss = (DateTime.Now - dt).TotalMilliseconds;
            ss = ss == 0 ? 1 : ss;
            //暂时停止 
            if (ss < 1000)
            {
                System.Threading.Thread.Sleep(1000 - (int)ss);
            }
            Console.WriteLine("{0:F2}/s,最后数据时间:{1}", speed , LastData);
        }

        DateTime LastData = DateTime.MinValue;

        /// <summary>
        /// 接收订阅数据
        /// </summary>
        /// <param name="c"></param>
        /// <param name="v"></param>

        void ReceiveData(StackExchange.Redis.RedisChannel c, StackExchange.Redis.RedisValue v)
        {
            var sbc = v.ToString();
            string[] ssb = sbc.Split(';');
            if (ssb.Length > 1) {
                if (DateTime.TryParse(ssb[0], out LastData)){
                    System.Threading.Interlocked.Increment(ref llong); }

            }
            //var csub = subs[llong % subs.Count];
            ////联接断开时直接返回
            //if (!csub.IsConnected()) { return; }
            //csub.Subscribe(实时数据通知名, ReceiveData);
        }
    }
}

